package com.yonyou.rabbitmq.config;

import com.yonyou.rabbitmq.common.MessageConstants;
import com.yonyou.rabbitmq.common.enums.SiteEnum;
import com.yonyou.rabbitmq.producer.ExtCorrelationData;
import com.yonyou.rabbitmq.producer.MessageInfo;
import com.yonyou.rabbitmq.producer.MessageProcess;
import com.yonyou.rabbitmq.producer.ReliableRabbitTemplate;
import com.yonyou.rabbitmq.utils.MessageUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.policy.TimeoutRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import javax.annotation.Resource;
import java.util.Optional;

@Slf4j
@EnableRabbit
@Configuration
public class TemplateConfig {

    @Resource
    RabbitProperties rabbitProperties;

    public ConnectionFactory getConnectionFactory(RabbitProperties rabbitProperties, String vhost){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setAddresses(rabbitProperties.getAddresses());
        cachingConnectionFactory.setUsername(rabbitProperties.getUsername());
        cachingConnectionFactory.setPassword(rabbitProperties.getPassword());
        cachingConnectionFactory.setVirtualHost(vhost);
//        cachingConnectionFactory.setPublisherConfirms(rabbitProperties.getPublisherConfirmType());
        cachingConnectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns());
        return cachingConnectionFactory;
    }

    @Bean(MessageConstants.GF_CONNECTION_FACTORY)
    @Primary
    public ConnectionFactory gfConnectionFactory() {
        String vhost = SiteEnum.getValue(SiteEnum.GF.getKey());
        return this.getConnectionFactory(rabbitProperties,vhost);
    }

    @Bean(MessageConstants.GF_RABBIT_TEMPLATE)
    @Primary
    public RabbitTemplate gfRabbitTemplate(@Qualifier(MessageConstants.GF_CONNECTION_FACTORY) ConnectionFactory connectionFactory, MessageProcess messageProcess) {
        ReliableRabbitTemplate rabbitTemplate = new ReliableRabbitTemplate();
        initTemplate(rabbitTemplate,connectionFactory,messageProcess);
        return rabbitTemplate;
    }

    @Bean(MessageConstants.GF_RABBIT_ADMIN)
    @Primary
    public RabbitAdmin gfRabbitAdmin(@Qualifier(MessageConstants.GF_CONNECTION_FACTORY) ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    // 配置批量监听容器
    @Bean(MessageConstants.GF_CONTAINER_FACTORY)
    @Primary
    public SimpleRabbitListenerContainerFactory gfContainerFactory(@Qualifier(MessageConstants.GF_CONNECTION_FACTORY) ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        initContainer(connectionFactory, factory);
        return factory;
    }

    @Bean(MessageConstants.ZW_CONNECTION_FACTORY)
    public ConnectionFactory zwConnectionFactory() {
        String vhost = SiteEnum.getValue(SiteEnum.ZW.getKey());
        return this.getConnectionFactory(rabbitProperties,vhost);
    }

    @Bean(MessageConstants.ZW_RABBIT_TEMPLATE)
    public ReliableRabbitTemplate zwRabbitTemplate(@Qualifier(MessageConstants.ZW_CONNECTION_FACTORY) ConnectionFactory connectionFactory,MessageProcess messageProcess) {
        ReliableRabbitTemplate rabbitTemplate = new ReliableRabbitTemplate();
        initTemplate(rabbitTemplate,connectionFactory,messageProcess);
        return rabbitTemplate;
    }

    @Bean(MessageConstants.ZW_RABBIT_ADMIN)
    public RabbitAdmin zwRabbitAdmin(@Qualifier(MessageConstants.ZW_CONNECTION_FACTORY) ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public MessageRecoverer gfMessageRecoverer(@Qualifier(MessageConstants.GF_RABBIT_TEMPLATE) RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate,MessageConstants.DLX_EXCHANGE,MessageConstants.DLK_ROUTING_KEY);
    }

    @Bean
    public MessageRecoverer zwMessageRecoverer(@Qualifier(MessageConstants.ZW_RABBIT_TEMPLATE) RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate,MessageConstants.DLX_EXCHANGE,MessageConstants.DLK_ROUTING_KEY);
    }

    // 配置批量监听容器
    @Bean(MessageConstants.ZW_CONTAINER_FACTORY)
    public SimpleRabbitListenerContainerFactory zwContainerFactory(@Qualifier(MessageConstants.ZW_CONNECTION_FACTORY) ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        initContainer(connectionFactory, factory);
        return factory;
    }

    private void initContainer(ConnectionFactory connectionFactory, SimpleRabbitListenerContainerFactory factory) {
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //设置消费者的消息确认模式
        RabbitProperties.SimpleContainer simple = rabbitProperties.getListener().getSimple();
        factory.setAcknowledgeMode(simple.getAcknowledgeMode());
        factory.setPrefetchCount(simple.getPrefetch());
        factory.setConcurrentConsumers(simple.getConcurrency());  //设置线程数
        factory.setMaxConcurrentConsumers(simple.getMaxConcurrency()); //最大线程数
        //设置重试参数
        factory.setAdviceChain(getRetryOperationsInterceptor());
    }

    @Bean
    public RetryOperationsInterceptor getRetryOperationsInterceptor(){
        RabbitProperties.ListenerRetry retry = rabbitProperties.getListener().getSimple().getRetry();
        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(retry.getInitialInterval().toMillis());
        backOffPolicy.setMultiplier(retry.getMultiplier());
        backOffPolicy.setMaxInterval(retry.getMaxInterval().toMillis());
        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(retry.getMaxAttempts()));
        return RetryInterceptorBuilder.stateless().retryOperations(retryTemplate).recoverer(new RejectAndDontRequeueRecoverer()).build();
    }

    public void initTemplate(ReliableRabbitTemplate template, ConnectionFactory connectionFactory, MessageProcess messageProcess){
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        template.setConnectionFactory(connectionFactory);
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        template.setMandatory(true);
        template.setMessageProcess(messageProcess);

        RetryTemplate retryTemplate = new RetryTemplate();

        TimeoutRetryPolicy policy = new TimeoutRetryPolicy();
        policy.setTimeout(15000);
        retryTemplate.setRetryPolicy(Optional.ofNullable(messageProcess.getRetryPolicy()).orElse(policy));
        template.setRetryTemplate(retryTemplate);

        //确认回调
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if(null!=correlationData && correlationData instanceof ExtCorrelationData){
                ExtCorrelationData extData = (ExtCorrelationData) correlationData;

                MessageInfo messageInfo = extData.getMessageInfo();
                messageInfo.setFailReason(Optional.ofNullable(cause).orElse(""));

                if (ack) {
                    messageProcess.getMessageAck().accept(messageInfo);
                } else {
                    messageProcess.getMessageNack().accept(messageInfo);
                }
            }
        });

        //找不到路由回调
        template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            String messageBody = MessageUtils.getBodyStr(message);
            String messageProperties = MessageUtils.getPropertiesStr(message);
            messageProcess.getMessageRouteFail().accept(messageProperties, messageBody, exchange, routingKey);
        });

    }

}